package com.allyes.flume.interceptor.parser;

import java.io.File;
import java.io.IOException;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.io.DatumReader;

public class HdfsAvroFileOpen {

  public static void main(String[] args) {
    HdfsAvroFileOpen ins = new HdfsAvroFileOpen();

    try {
      ins.openAinsightRawlog("avro_container_file/rawlog.ainsight");
      ins.openAimRawlog("avro_container_file/rawlog.aim");
      ins.openAdxRawlog("avro_container_file/rawlog.adx");

    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  public void openAinsightRawlog(String filename) throws IOException {
    System.out.println("\nOpening ainsight rawlog: " + filename);
    File file = new File(filename);
    Schema schema = com.allyes.log.RawLog.getClassSchema();
    DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
    DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(file, datumReader);
    GenericRecord record = null;
    while (dataFileReader.hasNext()) {
      record = dataFileReader.next(record);
      System.out.println(record);
    }
    dataFileReader.close();
  }

  public void openAimRawlog(String filename) throws IOException {
    System.out.println("\nOpening aim rawlog: " + filename);
    File file = new File(filename);
    Schema schema = com.allyes.aim.log.RawLog.getClassSchema();
    DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
    DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(file, datumReader);
    GenericRecord record = null;
    while (dataFileReader.hasNext()) {
      record = dataFileReader.next(record);
      System.out.println(record);
    }
    dataFileReader.close();
  }

  public void openAdxRawlog(String filename) throws IOException {
    System.out.println("\nOpening adx rawlog: " + filename);
    File file = new File(filename);
    Schema schema = com.allyes.adx.log.RawLog.getClassSchema();
    DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
    DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(file, datumReader);
    GenericRecord record = null;
    while (dataFileReader.hasNext()) {
      record = dataFileReader.next(record);
      System.out.println(record);
    }
    dataFileReader.close();
  }

}
